# !/usr/bin/env python3
# coding=utf8
"""
"n分钟K线"生成器
"""
import datacompy
import datetime
import functools
import logging
import numpy
import pandas
import pathlib
from typing import Any, Dict, List, Set, Tuple, Type, Optional, Union, Callable

from vnpy.event import Event, EventEngine, EVENT_TIMER
from vnpy.trader.constant import Exchange, Interval
from vnpy.trader.object import BarData, TickData, TradeData

from vnpy_zxtools.chinese_mainland_time_config import ChineseMainlandTimeConfig


class BarGeneratorVnpy:
    """
    For:
    1. generating 1 minute bar data from tick data
    2. generating x minute bar/x hour bar data from 1 minute data
    Notice:
    1. for x minute bar, x must be able to divide 60: 2, 3, 5, 6, 10, 15, 20, 30
    2. for x hour bar, x can be any number
    要合成"n分钟K线", 就需要"1分钟K线", 并且, 这个类又要兼顾实时行情和回测数据,
    所以它的主要职责是合成"n分钟K线", 同时附带了一个根据实时行情合成"1分钟K线"的功能,
    """

    class BarSummary(object):
        """"""

        def __init__(self, tp_begin: datetime.datetime, tp_last: datetime.datetime, tp_end: datetime.datetime, business_day: datetime.date) -> None:
            """"""
            assert isinstance(business_day, datetime.date)
            # 例如 [02:29~09:06): 02:29, 09:00, 09:01, 09:02, 09:03, 09:04, 09:05,
            self.tp_begin: datetime.datetime = tp_begin  # 02:29
            self.tp_last: datetime.datetime = tp_last  # 09:05
            self.tp_end: datetime.datetime = tp_end  # 09:06 (疑似并非永远有: tp_last + 1minute => tp_end)
            self.business_day: datetime.date = business_day

    class BarSummaryCache(object):
        """"""

        def __init__(self, window: int) -> None:
            """"""
            assert isinstance(window, int) and 0 < window
            self.window: int = window  # 一旦初始化, 就不再改变了,
            self.business_day: datetime.date = None
            self.cache: Dict[datetime.datetime, BarGeneratorVnpy.BarSummary] = None  # 所有的有效的"1分钟K线"的开始时间戳,
            self.beg: datetime.datetime = None
            self.end: datetime.datetime = None

        def get_bar_summary(self, bar: BarData) -> Optional["BarGeneratorVnpy.BarSummary"]:
            """"""
            bar_dttm: datetime.datetime = bar.datetime.replace(tzinfo=None)

            self.initialize(exchange=bar.exchange, symbol=bar.symbol, dttm=bar_dttm)
            summary: Optional[BarGeneratorVnpy.BarSummary] = self.cache.get(bar_dttm, None)

            return summary

        def initialize(self, exchange: Exchange, symbol: str, dttm: datetime.datetime) -> None:
            """
            它兼具"初始化"和"重新初始化"的功能,
            """
            business_day: Optional[datetime.date] = None

            if (self.business_day is None) or not (self.beg <= dttm <= self.end):
                # 如果未初始化, 或, 时间戳不在时间段内, 则要重新初始化,
                business_day: Optional[datetime.date] = ChineseMainlandTimeConfig.guess_business_day(dttm)
                if business_day is None:
                    return

            if (business_day is not None) and (self.business_day != business_day):
                # tp => 命名取自 C++ 的 std::chrono::time_point
                tps: List[datetime.datetime] = ChineseMainlandTimeConfig.calc_for_business_day(
                    exchange=exchange,
                    symbol=symbol,
                    window=1,
                    is_left=True,
                    add_boundary_dttm=True,
                    business_day=business_day,
                    is_default=True,
                )

                end_tp: datetime.datetime = tps[-1]  # 最后一个时间戳不是有效的"1分钟K线"的时间戳, chunk 的时候应该去掉它,
                chunk_data: List[List[datetime.datetime]] = ChineseMainlandTimeConfig.list_chunk(tps[:-1], self.window)
                last_index: int = len(chunk_data) - 1

                cache: Dict[datetime.datetime, BarGeneratorVnpy.BarSummary] = {}

                for i in range(0, len(chunk_data), 1):
                    is_last: bool = (i == last_index)
                    curr: List[datetime.datetime] = chunk_data[i]

                    summary: BarGeneratorVnpy.BarSummary = BarGeneratorVnpy.BarSummary(
                        tp_begin=curr[0],
                        tp_last=curr[-1],
                        tp_end=end_tp if is_last else chunk_data[i + 1][0],
                        business_day=business_day,
                    )

                    for tp in curr:
                        cache[tp] = summary

                self_beg: datetime.datetime = tps[+0] - datetime.timedelta(hours=2)
                self_end: datetime.datetime = tps[-1] + datetime.timedelta(hours=2)
                # TODO: 这样的命令应该不是原子的吧?
                self.business_day, self.cache, self.beg, self.end = business_day, cache, self_beg, self_end

            pass

    def __init__(
        self,
        on_bar: Callable,
        window: int = 0,
        on_window_bar: Callable = None,
        interval: Interval = Interval.MINUTE,
        daily_end: datetime.time = None,
        is_tdx_datetime: bool = False,
    ) -> None:
        """Constructor"""
        """
        on_bar: 用Tick生成的"1分钟K线"的"回调函数"
        window: 生成的("n分钟K线")的n
        on_window_bar: 生成的("n分钟K线")的"回调函数"
        interval: (表示要)生成"n分钟K线"
        is_tdx_datetime: 使用通达信规则的时间戳
        """
        assert isinstance(window, int) and 0 < window

        # 用Tick生成的"1分钟K线"
        self.bar: BarData = None
        # 用Tick生成的"1分钟K线"的"回调函数"
        self.on_bar: Callable = on_bar

        # 生成"n分钟K线"还是"n小时K线"还是"1日K线"
        self.interval: Interval = interval
        # 因为代码逻辑只处理了它, 所以在这里做一个强检查,
        assert self.interval in (Interval.MINUTE,)

        # 用不到下面的代码, 遂注释掉它,
        # self.interval_count: int = 0
        # self.hour_bar: BarData = None
        # self.daily_bar: BarData = None

        # 生成的("n分钟K线"/"n小时K线")的n
        self.window: int = window
        # 生成的("n分钟K线"/"n小时K线")
        self.window_bar: BarData = None
        # 生成的("n分钟K线"/"n小时K线"/"1日K线")的"回调函数"
        self.on_window_bar: Callable = on_window_bar

        self.last_tick: TickData = None

        # 用不到下面的代码, 遂注释掉它,
        # self.daily_end: time = daily_end
        # if self.interval == Interval.DAILY and not self.daily_end:
        #     raise RuntimeError("合成日K线必须传入每日收盘时间")

        self.is_tdx_datetime: bool = is_tdx_datetime

        self.bsc: BarGeneratorVnpy.BarSummaryCache = BarGeneratorVnpy.BarSummaryCache(window=self.window)

        self.event_engine: EventEngine = None

        self.export_window_bar_flag: bool = False
        self.export_window_bar_bars: List[BarData] = []

    def __del__(self):
        """"""
        if self.event_engine:
            self.event_engine.unregister(EVENT_TIMER, self.process_timer_event)
            self.event_engine = None

    def register_timer_event(self, event_engine: EventEngine) -> None:
        """"""
        if self.event_engine:
            assert self.event_engine == event_engine
            return

        self.event_engine: EventEngine = event_engine
        self.event_engine.register(EVENT_TIMER, self.process_timer_event)

    def update_tick(self, tick: TickData) -> None:
        """
        Update new tick data into generator.
        if tick有问题, then 丢弃,
        if tick在另一分钟, then (回调self.bar, 并用tick构造新的self.bar)
        else (将tick合入self.bar, 并更新self.last_tick)

        摘自 vnpy_ctastrategy/strategies/boll_channel_strategy.py 的一个用法示例:

        class BollChannelStrategy(CtaTemplate):

            def __init__(self, cta_engine, strategy_name, vt_symbol, setting):
                super().__init__(cta_engine, strategy_name, vt_symbol, setting)
                self.bg = BarGenerator(on_bar=self.on_bar, window=15, on_window_bar=self.on_15min_bar, interval=Interval.MINUTE)

            def on_tick(self, tick: TickData):
                self.bg.update_tick(tick)

            def on_bar(self, bar: BarData):
                self.bg.update_bar(bar)

            def on_15min_bar(self, bar: BarData):
                pass

        BarGenerator.update_tick(tick)  <=> BarGenerator.[on_bar]
                                        <=> Strategy.on_bar
                                        <=> BarGenerator.update_bar <=> BarGenerator.[on_window_bar]
                                                                    <=> Strategy.on_15min_bar
        """
        new_minute: bool = False

        # Filter tick data with 0 last price
        if not tick.last_price:
            # TODO: 一个股票很不活跃, 开盘好几分钟了, 一笔成交都没有, 此时应当有K线,
            return

        # Filter tick data with older timestamp
        if self.last_tick and tick.datetime < self.last_tick.datetime:
            return

        if not self.bar:
            new_minute = True
        elif (
            (self.bar.datetime.minute != tick.datetime.minute)
            or (self.bar.datetime.hour != tick.datetime.hour)
        ):
            # 用Tick驱动"1分钟K线"的"回调函数",
            self.bar.datetime = self.bar.datetime.replace(
                second=0, microsecond=0
            )
            self.on_bar(self.bar)

            new_minute = True

        if new_minute:
            # 在每个1分钟的开始, 用Tick生成"1分钟K线",
            self.bar = BarData(
                gateway_name=tick.gateway_name,
                symbol=tick.symbol,
                exchange=tick.exchange,
                datetime=tick.datetime,
                interval=Interval.MINUTE,
                volume=0,
                turnover=0,
                open_interest=tick.open_interest,
                open_price=tick.last_price,
                high_price=tick.last_price,
                low_price=tick.last_price,
                close_price=tick.last_price,
            )
        else:
            self.bar.high_price = max(self.bar.high_price, tick.last_price)
            if tick.high_price > self.last_tick.high_price:
                self.bar.high_price = max(self.bar.high_price, tick.high_price)

            self.bar.low_price = min(self.bar.low_price, tick.last_price)
            if tick.low_price < self.last_tick.low_price:
                self.bar.low_price = min(self.bar.low_price, tick.low_price)

            self.bar.close_price = tick.last_price
            self.bar.open_interest = tick.open_interest
            # TODO: 为什么要这样做?
            self.bar.datetime = tick.datetime

        if self.last_tick:
            volume_change: float = tick.volume - self.last_tick.volume
            # 这个"1分钟K线"区间内的成交量
            self.bar.volume += max(volume_change, 0)

            turnover_change: float = tick.turnover - self.last_tick.turnover
            self.bar.turnover += max(turnover_change, 0)

        self.last_tick = tick

    def update_bar(self, bar: BarData) -> None:
        """
        Update 1 minute bar into generator
        """
        # 因为代码逻辑只处理了它, 所以在这里做一个强检查,
        assert self.interval in (Interval.MINUTE,)

        if True:
            self.update_bar_minute_window(bar)

    def update_bar_minute_window_deprecated(self, bar: BarData) -> None:
        """"""
        # C++14 的一个关键字 deprecated
        # 用"1分钟K线"计算"n分钟K线"(n=window), 如果"n分钟K线"计算完成了, 就回调,
        # If not inited, create window bar object
        if not self.window_bar:
            dt: datetime = bar.datetime.replace(second=0, microsecond=0)
            self.window_bar = BarData(
                symbol=bar.symbol,
                exchange=bar.exchange,
                datetime=dt,
                gateway_name=bar.gateway_name,
                open_price=bar.open_price,
                high_price=bar.high_price,
                low_price=bar.low_price
            )
        # Otherwise, update high/low price into window bar
        else:
            self.window_bar.high_price = max(
                self.window_bar.high_price,
                bar.high_price
            )
            self.window_bar.low_price = min(
                self.window_bar.low_price,
                bar.low_price
            )

        # Update close price/volume/turnover into window bar
        self.window_bar.close_price = bar.close_price
        self.window_bar.volume += bar.volume
        self.window_bar.turnover += bar.turnover
        self.window_bar.open_interest = bar.open_interest

        # Check if window bar completed
        if not (bar.datetime.minute + 1) % self.window:
            self.on_window_bar(self.window_bar)
            self.window_bar = None

    def generate(self) -> Optional[BarData]:
        """
        Generate the bar data and call callback immediately.
        提问前请先看：vn.py使用FAQ - 主题 - VeighNa量化社区
        https://www.vnpy.com/forum/topic/250-ti-wen-qian-qing-xian-kan-:vn-pyshi-yong-faq
        Q：郑商所的品种都收不到14:59这一根bar
        A：郑商所的数据推送，没有3点后的最后一个tick用于标识收盘完成，所以要调用BarGenerator.generate函数来做最终的强制K线生成
        """
        bar: BarData = self.bar

        if self.bar:
            bar.datetime = bar.datetime.replace(second=0, microsecond=0)
            self.on_bar(bar)

        self.bar = None
        return bar

    def process_timer_event(self, event: Event) -> None:
        """"""
        # now: datetime.datetime = datetime.datetime.now()

        # if now.hour == 15 and now.minute == 0 and 30 <= now.second < 60:
        #     assert (event.type == EVENT_TIMER) and (event.data is None)

        #     bar: BarData = self.bar
        #     if bar and bar.exchange == Exchange.CZCE:  # 郑商所
        #         self.generate()
        # 因为它一般会被管理在 RecorderEngine/CtaEngine 下, 暂定让它们触发 generate 函数,
        pass

    def on_window_bar_callable(self, bar: BarData) -> None:
        """"""
        self.on_window_bar(bar)

        if self.export_window_bar_flag:
            filename: str = f"{self.__class__.__name__}_{bar.exchange.value}_{bar.symbol}.csv"
            filepath: pathlib.Path = pathlib.Path(__file__).parent.joinpath(filename)
            # TODO: 如果已经存在旧文件, 那么就追加到旧文件上了,
            with open(file=filepath, mode="a", encoding="utf8") as f:
                f.write(BarGeneratorVectorized.BarData_to_line(bar=bar))

    def update_bar_minute_window(self, bar: BarData) -> None:
        """"""
        summary: Optional[BarGeneratorVnpy.BarSummary] = self.bsc.get_bar_summary(bar)
        if not summary:
            logging.warning(f"skip bar={bar}")
            return
        self.deliver_business_day(dst=bar, src=summary)

        # summary 对应的"n分钟K线"的时间戳
        summary_window_bar_datetime: datetime.datetime = summary.tp_end if self.is_tdx_datetime else summary.tp_begin
        summary_window_bar_datetime: datetime.datetime = summary_window_bar_datetime.replace(tzinfo=bar.datetime.tzinfo)

        # If not inited, create window bar object
        if not self.window_bar:
            # 没有"n分钟K线", 收到的"1分钟K线"构造"n分钟K线",
            self.window_bar = BarData(
                gateway_name=bar.gateway_name,
                symbol=bar.symbol,
                exchange=bar.exchange,
                datetime=summary_window_bar_datetime,
                interval=None,
                volume=0,
                turnover=0,
                open_interest=0,
                open_price=bar.open_price,
                high_price=bar.high_price,
                low_price=bar.low_price,
                close_price=0,
            )
            self.deliver_business_day(dst=self.window_bar, src=summary)

        if self.window_bar.datetime == summary_window_bar_datetime:
            # 存在"n分钟K线", 收到的"1分钟K线"属于"n分钟K线", 合入,
            self.window_bar.high_price = max(
                self.window_bar.high_price,
                bar.high_price
            )
            self.window_bar.low_price = min(
                self.window_bar.low_price,
                bar.low_price
            )

            # Update close price/volume/turnover into window bar
            self.window_bar.close_price = bar.close_price
            self.window_bar.volume += bar.volume
            self.window_bar.turnover += bar.turnover
            self.window_bar.open_interest = bar.open_interest

            if bar.datetime == summary.tp_last.replace(tzinfo=bar.datetime.tzinfo):
                # 收到的"1分钟K线"属于"n分钟K线"的最后一根,
                self.on_window_bar_callable(self.window_bar)
                self.window_bar = None

        else:
            # 存在"n分钟K线", 收到的"1分钟K线"不属于"n分钟K线", 回调+构造,
            self.on_window_bar_callable(self.window_bar)

            self.window_bar = BarData(
                gateway_name=bar.gateway_name,
                symbol=bar.symbol,
                exchange=bar.exchange,
                datetime=summary_window_bar_datetime,
                interval=None,
                volume=bar.volume,
                turnover=bar.turnover,
                open_interest=bar.open_interest,
                open_price=bar.open_price,
                high_price=bar.high_price,
                low_price=bar.low_price,
                close_price=bar.close_price,
            )
            self.deliver_business_day(dst=self.window_bar, src=summary)

    @classmethod
    def deliver_business_day(cls, dst, src) -> None:
        """
        传递 business_day 属性(如果 src 有 business_day 就将其赋值给 dst)
        像这种"一个对象原本没有这个属性, 因为一些需要, 强行赋值给它"的情况, 为了方便追溯, 统一用函数进行赋值,
        """
        if hasattr(src, "business_day"):
            dst.business_day = src.business_day

    @classmethod
    def business_day(cls, src: Union[BarData, TradeData]) -> datetime.date:
        """"""
        if hasattr(src, "business_day"):
            return src.business_day
        else:
            return src.datetime.date()


def aaabbb(
    exchange: Exchange,
    symbol: str,
    interval: Interval,
    start: datetime,
    end: datetime,
) -> List[BarData]:
    """"""
    from vnpy.trader.database import get_database

    database = get_database()

    bars: List[BarData] = database.load_bar_data(
        symbol=symbol,
        exchange=exchange,
        interval=interval,
        start=start,
        end=end,
    )

    df: pandas.DataFrame = pandas.DataFrame()

    if bars:
        for bar in bars:
            bar.datetime = bar.datetime.strftime("%Y-%m-%d %H:%M:%S")
            bar.exchange = bar.exchange.value
            bar.interval = bar.interval.value

        df: pandas.DataFrame = pandas.DataFrame(data=[bar.__dict__ for bar in bars])

        df.drop(columns="gateway_name", inplace=True)

        df["datetime"] = pandas.to_datetime(df["datetime"])  # 字符串转时间戳

        # inplace: 是否在原 DataFrame 上修改, 默认为 False
        # verify_integrity: 是否检查索引有无重复, 默认为 False, 若设置为 True 会影响程序性能,
        df.set_index("datetime", inplace=True, verify_integrity=False)

    return df


def xxxxx(
    exchange: Exchange,
    symbol: str,
    interval: Interval,
    start: datetime.datetime,
    end: datetime.datetime,
    window: int,
    size: int,
    is_tdx_datetime: bool = False,
) -> pandas.DataFrame:
    """
    window: 生成的("n分钟K线")的n,
    size: 需要预加载多少个"n分钟K线", 参考 vnpy/trader/utility.py 的 ArrayManager 的 size 参数,
    is_tdx_datetime: 使用通达信的时间戳,
    """
    df0: pandas.DataFrame = aaabbb(
        exchange=exchange,
        symbol=symbol,
        interval=interval,
        start=start,
        end=end,
    )

    window_bar_s: List[BarData] = []  # 生成的"n分钟K线"都存到这里

    def on_window_bar_callable(container: List[BarData], bar: BarData):
        """"""
        assert bar.interval is None
        bar.interval = Interval.MINUTE
        container.append(bar)

    func: callable = functools.partial(on_window_bar_callable, window_bar_s)

    bg: BarGeneratorVnpy = BarGeneratorVnpy(
        on_bar=None,
        window=window,
        on_window_bar=None,
        interval=interval,
        is_tdx_datetime=is_tdx_datetime,
    )
    bg.on_bar = bg.update_bar
    bg.on_window_bar = func

    for idx, row in df0.iterrows():
        bar: BarData = BarData(
            gateway_name="placeholder",
            symbol=row.symbol,
            exchange=Exchange(row.exchange),
            datetime=idx.to_pydatetime(),
            interval=Interval(row.interval),
            volume=row.volume,
            turnover=row.turnover,
            open_interest=row.open_interest,
            open_price=row.open_price,
            high_price=row.high_price,
            low_price=row.low_price,
            close_price=row.close_price,
        )
        bg.update_bar(bar=bar)

    data: list = []

    for idx, bar in enumerate(window_bar_s):
        bar: BarData = bar

        d: dict = bar.__dict__
        d["exchange"] = d["exchange"].value
        d["interval"] = d["interval"].value
        d.pop("gateway_name")
        d.pop("vt_symbol")
        data.append(d)

    df1: pandas.DataFrame = pandas.DataFrame(data=data)
    df1.set_index("datetime", inplace=True, verify_integrity=False)

    return df1


class BarGeneratorVectorized(object):
    """"""

    def __init__(
        self,
        exchange: Exchange,
        symbol: str,
        interval: Interval,
        start: datetime.datetime,
        end: datetime.datetime,
        window: int,
        size: int,
        is_tdx_datetime: bool = False,
    ) -> None:
        """"""
        self.exchange: Exchange = exchange
        self.symbol: str = symbol
        self.interval: Interval = interval
        self.start: datetime.datetime = start
        self.end: datetime.datetime = end
        self.window: int = window
        self.is_tdx_datetime: bool = is_tdx_datetime

        self.df_raw: pandas.DataFrame = None
        self.df_dst: pandas.DataFrame = None
        self.prices: Dict[datetime.datetime, float] = {}

    def execute(self) -> pandas.DataFrame:
        """"""
        self.df_raw: pandas.DataFrame = aaabbb(
            exchange=self.exchange,
            symbol=self.symbol,
            interval=self.interval,
            start=self.start,
            end=self.end,
        )

        beg = self.df_raw.index.min().to_pydatetime().date()
        end = self.df_raw.index.max().to_pydatetime().date()

        bins_window: List[datetime.datetime] = ChineseMainlandTimeConfig.calc_for_days(
            exchange=self.exchange,
            symbol=self.symbol,
            window=self.window,
            is_left=True,
            add_boundary_dttm=True,
            is_default=True,
            start_date=beg,
            end_date=end,
        )

        self.df_dst = self.gen_bar_in_window_minutes(df0=self.df_raw, bins=bins_window, is_tdx_datetime=self.is_tdx_datetime)  # noqa E501

        if False:
            self.save_to_file(df=self.df_dst)

        # 计算日线, 为了填充 self.prices 变量, 开始,

        bins_24h: List[datetime.datetime] = ChineseMainlandTimeConfig.calc_for_days(
            exchange=self.exchange,
            symbol=self.symbol,
            window=24 * 60,
            is_left=True,
            add_boundary_dttm=True,
            is_default=True,
            start_date=beg,
            end_date=end,
        )

        df_24h = self.gen_bar_in_window_minutes(df0=self.df_raw, bins=bins_24h, is_tdx_datetime=True)

        for idx, row in df_24h.iterrows():
            dttm: datetime.datetime = idx.to_pydatetime()
            if dttm < self.start:  # TODO: 不够优雅,
                continue
            close_price = row.close_price
            self.prices[dttm] = float(close_price)

        # 计算日线, 为了填充 self.prices 变量, 结束,

        return self.df_dst

    def gen_bar_in_window_minutes(cls, df0: pandas.DataFrame, bins: List[datetime.datetime], is_tdx_datetime: bool):
        """"""
        df0: pandas.DataFrame = df0.copy(deep=False)

        # assert isinstance(df0.index, pandas.DatetimeIndex)
        # assert isinstance(df0.index[0], pandas.Timestamp)
        # array_datetime = df0.index.to_pydatetime()
        # assert isinstance(array_datetime, numpy.ndarray)
        # assert isinstance(array_datetime[0], datetime.datetime)

        df0["datetime"] = df0.index
        df0["business_day"] = numpy.vectorize(ChineseMainlandTimeConfig.guess_business_day)(df0.index.to_pydatetime())

        # assert isinstance(df0["datetime"], pandas.Series)
        # assert isinstance(df0["datetime"][0], pandas.Timestamp)

        df1: pandas.DataFrame = df0.groupby(pandas.cut(df0.index, bins=bins, right=False)).aggregate(
            {
                "symbol": "first",
                "exchange": "first",
                "datetime": "first",
                "interval": "first",
                "volume": "sum",
                "turnover": "sum",
                "open_interest": "last",  # 持仓量,
                "open_price": "first",
                "high_price": "max",
                "low_price": "min",
                "close_price": "last",
                "business_day": "last",
            }
        ).dropna()

        if True:
            assert isinstance(df1.index, pandas.CategoricalIndex)
            assert isinstance(df1.index[0], pandas.Interval)
            # pandas.Interval.left
            # pandas.Interval.right
            df1["vnpy_datetime"] = numpy.vectorize(lambda _: _.left)(df1.index)
            df1["tdx_datetime"] = numpy.vectorize(lambda _: _.right)(df1.index)
            if is_tdx_datetime:
                df1["datetime"] = df1["tdx_datetime"]
            else:
                df1["datetime"] = df1["vnpy_datetime"]
            df1.drop(labels=["vnpy_datetime", "tdx_datetime"], axis=1, inplace=True)

        df1.reset_index(drop=True, inplace=True)  # drop=True 避免将旧索引添加为列,

        df1.set_index("datetime", inplace=True, verify_integrity=False)

        return df1

    @classmethod
    def BarData_to_line(cls, bar: BarData, delimiter: str = "\n") -> str:
        """"""
        # bar.gateway_name
        # bar.interval
        line: str = \
            f"{bar.symbol},{bar.exchange.value}," \
            f"{bar.datetime.strftime('%Y-%m-%d %H:%M:%S')}," \
            f"{bar.volume},{bar.turnover},{bar.open_interest}," \
            f"{bar.open_price},{bar.high_price},{bar.low_price},{bar.close_price}{delimiter}"
        return line

    @classmethod
    def save_to_file(cls, df: pandas.DataFrame) -> None:
        """"""
        bars: List[BarData] = []

        for idx, row in df.iterrows():
            idx: pandas.Timestamp = idx

            bar: BarData = BarData(
                gateway_name="pandas",
                symbol=row.symbol,
                exchange=Exchange(row.exchange),
                datetime=idx.to_pydatetime(),
                interval=Interval(row.interval),
                volume=row.volume,
                turnover=row.turnover,
                open_interest=row.open_interest,
                open_price=row.open_price,
                high_price=row.high_price,
                low_price=row.low_price,
                close_price=row.close_price,
            )
            bars.append(bar)

        lines: List[str] = [cls.BarData_to_line(bar) for bar in bars]

        filename: str = f"{cls.__name__}_{bar.exchange.value}_{bar.symbol}.csv"
        filepath: pathlib.Path = pathlib.Path(__file__).parent.joinpath(filename)
        with open(file=filepath, mode="w", encoding="utf8") as f:
            f.writelines(lines)

    @classmethod
    def execute_one(
        cls,
        exchange: Exchange,
        symbol: str,
        interval: Interval,
        start: datetime.datetime,
        end: datetime.datetime,
        window: int,
        size: int,
        is_tdx_datetime: bool = False,
    ) -> pandas.DataFrame:
        """
        window: 生成的("n分钟K线")的n,
        size: 需要预加载多少个"n分钟K线", 参考 vnpy/trader/utility.py 的 ArrayManager 的 size 参数,
        is_tdx_datetime: 使用通达信规范的时间戳, 方便对比通达信的数据,
        """
        df0: pandas.DataFrame = aaabbb(
            exchange=exchange,
            symbol=symbol,
            interval=interval,
            start=start,
            end=end,
        )

        # assert isinstance(df0.index, pandas.DatetimeIndex)
        # assert isinstance(df0.index[0], pandas.Timestamp)
        # array_datetime = df0.index.to_pydatetime()
        # assert isinstance(array_datetime, numpy.ndarray)
        # assert isinstance(array_datetime[0], datetime.datetime)

        df0["datetime"] = df0.index

        # assert isinstance(df0["datetime"], pandas.Series)
        # assert isinstance(df0["datetime"][0], pandas.Timestamp)

        beg = df0.index.min().to_pydatetime().replace(hour=0, minute=0, second=0, microsecond=0)
        end = df0.index.max().to_pydatetime().replace(hour=0, minute=0, second=0, microsecond=0)

        bins: List[datetime.datetime] = ChineseMainlandTimeConfig.calc_for_days(
            exchange=exchange,
            symbol=symbol,
            window=window,
            is_left=True,
            add_boundary_dttm=True,
            is_default=True,
            start_date=beg,
            end_date=end,
        )

        df1: pandas.DataFrame = df0.groupby(pandas.cut(df0.index, bins=bins, right=False)).aggregate(
            {
                "symbol": "first",
                "exchange": "first",
                "datetime": "first",
                "interval": "first",
                "volume": "sum",
                "turnover": "sum",
                "open_interest": "last",  # 持仓量,
                "open_price": "first",
                "high_price": "max",
                "low_price": "min",
                "close_price": "last",
            }
        ).dropna()

        if True:
            assert isinstance(df1.index, pandas.CategoricalIndex)
            assert isinstance(df1.index[0], pandas.Interval)
            # pandas.Interval.left
            # pandas.Interval.right
            df1["vnpy_datetime"] = numpy.vectorize(lambda _: _.left)(df1.index)
            df1["tdx_datetime"] = numpy.vectorize(lambda _: _.right)(df1.index)
            if is_tdx_datetime:
                df1["datetime"] = df1["tdx_datetime"]
            else:
                df1["datetime"] = df1["vnpy_datetime"]
            df1.drop(labels=["vnpy_datetime", "tdx_datetime"], axis=1, inplace=True)

        df1.reset_index(drop=True, inplace=True)  # drop=True 避免将旧索引添加为列,

        df1.set_index("datetime", inplace=True, verify_integrity=False)

        return df1


if __name__ == "__main__":
    exchange: Exchange = Exchange.SHFE
    symbol: str = "AUL8"
    interval: Interval = Interval.MINUTE
    start: datetime.datetime = datetime.datetime(year=2022, month=8, day=1)
    end: datetime.datetime = datetime.datetime(year=2022, month=12, day=1)
    window: int = 7
    size: int = 100
    is_tdx_datetime: bool = True

    df1: pandas.DataFrame = xxxxx(
        exchange=exchange,
        symbol=symbol,
        interval=interval,
        start=start,
        end=end,
        window=window,
        size=size,
        is_tdx_datetime=is_tdx_datetime,
    )

    df2: pandas.DataFrame = BarGeneratorVectorized.execute_one(
        exchange=exchange,
        symbol=symbol,
        interval=interval,
        start=start,
        end=end,
        window=window,
        size=size,
        is_tdx_datetime=True,
    )

    compare: datacompy.Compare = datacompy.Compare(df1, df2, on_index=True, df1_name="df1_vnpy", df2_name="df2_pandas")

    print(compare.report(sample_count=100, column_count=20))

    if False:
        pandas.set_option('display.max_rows', None)  # 显示所有行

        print("+" * 40, f"compare.df1_unq_rows 有 {compare.df1_unq_rows.shape[0]} 行")
        print(compare.df1_unq_rows)

        print("+" * 40, f"compare.df2_unq_rows 有 {compare.df2_unq_rows.shape[0]} 行")
        print(compare.df2_unq_rows)

# TODO: 用通达信的5分钟K线+日线 校验 n分钟K线的正确性
